package test.top2.com.atguigu.app.dws;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import test.top2.com.atguigu.app.fun.SplitFunction;
import test.top2.com.atguigu.bean.KeywordBean;
import test.top2.com.atguigu.utils.MyClickHouseUtil;
import test.top2.com.atguigu.utils.MyKafkaUtil;

/**
 * ClassName: Dws01_TrafficKeywordPageViewWindow
 * Package: test.top2.com.atguigu.app.dws
 * Description:
 *
 * @Author ChenJun(有志男青年)
 * @Create 2023/5/11 9:56
 * @Version 1.0
 */
public class Dws01_TrafficKeywordPageViewWindow {
    public static void main(String[] args) throws Exception {

        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

//        env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L);
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
//        env.getCheckpointConfig().enableExternalizedCheckpoints(
//                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
//        );
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(
//                10, Time.of(1L, TimeUnit.DAYS), Time.of(3L, TimeUnit.MINUTES)
//        ));
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
//        System.setProperty("HADOOP_USER_NAME", "atguigu");

        //TODO 2.使用FlinkSQL方式读取 DWD层 页面主题数据,同时提取时间戳生成WaterMark
        tableEnv.executeSql("" +
                "create table page_view(\n" +
                "`common` map<String>,String>,\n" +
                "`page` map<String,String>,\n" +
                "`ts` bigint,\n" +
                "`rt` as TO_TIMESTAMP_LTZ(ts,3)\n" +
                " WATERMARK FOR rt AS rt - INTERVAL '2' SECOND\n" +
                ")" + MyKafkaUtil.getKafkaSourceConnOption("dwd_traffic_page_log","keyword_page_view_221109"));

        //TODO 3.过滤出搜索数据
        Table filterTable = tableEnv.sqlQuery("" +
                "select \n" +
                "page['item'] item,\n" +
                "`rt`\n" +
                "from page_view\n" +
                "where page['item'] is not null\n" +
                "and page['last_page_id'] = 'search'\n" +
                "and page['item_type'] = 'keyword");

        tableEnv.createTemporaryView("filter_table", filterTable);

        //TODO 4.注册UDTF并分词
        tableEnv.createTemporarySystemFunction("split_fun", SplitFunction.class);
        Table splitTable = tableEnv.sqlQuery("" +
                " SELECT\n" +
                "  word,\n" +
                "  `rt`\n" +
                " FROM filter_table, LATERAL TABLE(split_fun(item))");
        tableEnv.createTemporaryView("split_table", splitTable);

        //TODO 5.分组、开窗、聚合
        Table resultTable = tableEnv.sqlQuery("" +
                "SELECT \n" +
                "   window_start, \n" +
                "   window_end,\n" +
                "   word, \n" +
                "   count(*) ct\n" +
                "FROM TABLE(\n" +
                "  TUMBLE(TABLE splitTable, DESCRIPTOR(rt), INTERVAL '10' seconds))\n" +
                "GROUP BY word,window_start, window_end");

        //TODO 6.将动态表转换为流写出到ClickHouse
        DataStream<KeywordBean> rowDataStream = tableEnv.toAppendStream(resultTable, KeywordBean.class);
        rowDataStream.print("rowDataStream-->");
        rowDataStream.addSink(MyClickHouseUtil.getSinkFunction("insert into dws_traffic_keyword_page_view_window " +
                "values(?,?,?,?,?)"));

        //TODO 7.启动任务
        env.execute("Dws01_TrafficKeywordPageViewWindow");
    }
}
